package com.ustcinfo.study.hbase.xujianan;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.*;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.client.replication.ReplicationAdmin;
import org.apache.hadoop.hbase.util.Bytes;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.sql.Connection;
import java.sql.SQLException;
import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.Map;

import static com.ustcinfo.ishare.bdp.hbase.HBaseOperationDao.putList;

public class HbaseConnectionUtil {
    private static Logger logger = LoggerFactory.getLogger(HbaseConnectionUtil.class);

    /*集群配置对象*/
    private static Configuration conf = null;

    /*集群连接对象*/
    private static Connection connection = null;

    /*hbase集群管理对象*/
    private static Admin admin = null;

    /*hbase replication管理对象*/
    private static ReplicationAdmin replicationAdmin = null;

    /**
     * 配置初始化
     */
    static {
        System.setProperty("hadoop.home.dir", "C:\\hadoop");
        conf = HBaseConfiguration.create();
        conf.set("hbase.zookeeper.property.clientPort", "2181");
        conf.set("hbase.zookeeper.quorum", "node1");
        try {
            connection = ConnectionFactory.createConnection(conf);
            admin = connection.getAdmin();
            replicationAdmin = new ReplicationAdmin(conf);
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /**
     * 判断表是否存在
     *
     * @param name
     * @return
     */
    public static boolean tableExist(String name) {
        try {
            return admin.tableExists(TableName.valueOf(name));
        } catch (IOException e) {
            e.printStackTrace();
            return false;
        }
    }

    /**
     * 创建表
     *
     * @param name
     * @param families
     */
    public static void createTable(String name, String... families) {
        TableName tableName = TableName.valueOf(name);
        HTableDescriptor hTableDescriptor = new HTableDescriptor(tableName);
        if (!tableExist(name)) {
            for (String family : families) {
                HColumnDescriptor hColumnDescriptor = new HColumnDescriptor(family);
                hTableDescriptor.addFamily(hColumnDescriptor);
            }
            try {
                admin.createTable(hTableDescriptor);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }


    /**
     * 批量插入
     *
     * @param name
     * @param puts
     */
    public static void putList(String name, List<Put> puts) {
        TableName tableName = TableName.valueOf(name);
        try (Table table = connection.getTable(tableName)) {
            for (Put put : puts) {
                table.put(put);
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    /*
     * 批量从mysql--->hbase （封装putList）
     * */
    public static void putListString(String sql, String col_name, String family) throws SQLException {
        List<Put> putLists = new ArrayList<>();
        Map<Integer, String> map = JDBCUtil.getStrMap(sql);

        logger.info("MySQL获取数据>>>>>" + map.toString());
//遍历key  表的id值
        for (Integer i : map.keySet()) {
            Put put = new Put(Integer.toString(i).getBytes());
            String value = map.get(i);
            logger.info(family);

            logger.info("value>>>>>>>" + value);
            put.addColumn(family.getBytes(), col_name.getBytes(), value.getBytes());
            logger.info("put对象>>>>" + put.toString());
            putLists.add(put);
        }
//        logger.info("putLists>>>>>"+putLists);
        putList("BD_TEST", putLists);
    }

    public static void main(String[] args) throws SQLException {
//        F1
        String sql_id = "select id,id from transfer_task_to";
        String sql_task_id = "select id,task_id from transfer_task_to";
        String sql_create_time = "select id,create_time from transfer_task_to";  //
        String sql_start_time = "select id,start_time from transfer_task_to";   //
        String sql_out_filter_number = "select id,out_filter_number from transfer_task_to";
        String sql_from_run_param = "select id,from_run_param  from transfer_task_to";
//        F2
        String sql_current_retry_times = "select id,current_retry_times from transfer_task_to";
        String sql_retry_flag = "select id,retry_flag from transfer_task_to";
        String sql_parent_task_id = "select id,parent_task_id from transfer_task_to";
        String sql_end_time = "select id,end_time from transfer_task_to";  //
        String sql_kill_time = "select id,kill_time from transfer_task_to";  //
        String sql_failure_time = "select id,failure_time from transfer_task_to"; //


//        创建表 “BD_TEST”
//        createTable("BD_TEST", "F1,F2");
//F1  数据导入
        putListString(sql_id, "id", "F1");
        putListString(sql_task_id, "task_id", "F1");
        putListString(sql_create_time, "create_time", "F1"); //
        putListString(sql_start_time, "start_time", "F1");  //
        putListString(sql_out_filter_number, "out_filter_number", "F1");
        putListString(sql_from_run_param, "from_run_param", "F1");

//F2 数据导入
        putListString(sql_current_retry_times, "current_retry_times", "F2");
        putListString(sql_retry_flag, "retry_flag", "F2");
        putListString(sql_parent_task_id, "parent_task_id", "F2");
        putListString(sql_end_time, "end_time", "F2");
        putListString(sql_kill_time, "kill_time", "F2");
        putListString(sql_failure_time, "failure_time", "F2");


    }
}
